package com.ryan.mdp.reference;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageFormatException;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TemporaryQueue;

import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.jxpath.JXPathContext;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.connection.JmsResourceHolder;
import org.springframework.jms.remoting.JmsInvokerClientInterceptor;
import org.springframework.jms.support.JmsUtils;
import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.jms.support.destination.DynamicDestinationResolver;
import org.springframework.remoting.RemoteAccessException;
import org.springframework.remoting.RemoteInvocationFailureException;
import org.springframework.remoting.support.RemoteInvocation;
import org.springframework.util.ClassUtils;

import com.ryan.mdp.annotation.MessageProperty;
import com.ryan.mdp.annotation.parser.MdpMethodConfig;
import com.ryan.mdp.annotation.parser.MdpServiceConfig;
import com.ryan.mdp.callback.DefaultMdpCallback;
import com.ryan.mdp.callback.MdpCallback;
import com.ryan.mdp.constants.MessagePropertiesConstants;
import com.ryan.mdp.exception.MdpException;
import com.ryan.mdp.exception.MdpMessages;
import com.ryan.mdp.json.MdpInvocationResult;

public class MdpInvokerClientInterceptor extends JmsInvokerClientInterceptor implements MethodInterceptor,
        ExceptionListener {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final DestinationResolver destinationResolver = new DynamicDestinationResolver();

    private String[] destinations;
    private boolean pubSubDomain;
    private String concurrency;
    private boolean transacted = false;
    private String source;

    /**
     * 来自于xml配置文件中的值，如果此值大于默认值0，将覆盖该interface中所有method上的MessageProperty
     * Annotation中设置的值
     */
    private long timeout;
    private MessageConverter messageConverter = null;
    private Connection connection = null;
    private TemporaryQueue responseQueue = null;
    private List<Session> sessions;
    private List<MessageConsumer> consumers;

    private final AtomicLong correlationIDGenerator = new AtomicLong(0);
    // 应答桶
    private final ConcurrentMap<Long, ResponseTub> responseTubs = new ConcurrentHashMap<Long, ResponseTub>();
    private MdpServiceConfig mdpServiceConfig;
    private String hostName;
    private String hostIP;

    @Override
    public Object invoke(MethodInvocation methodInvocation) throws Throwable {
        if (AopUtils.isToStringMethod(methodInvocation.getMethod())) {
            return "JMS invoker proxy for destination [" + StringUtils.join(this.destinations) + "]";
        }
        logger.info("invoking mdp service [{}]", methodInvocation.getMethod());
        RemoteInvocation invocation = createRemoteInvocation(methodInvocation);
        MdpMethodConfig methodConfig = mdpServiceConfig.get(methodInvocation.getMethod());
        MdpInvocationResult result = null;
        try {
            result = executeRequest(invocation, methodConfig);
        } catch (MdpException ex) {
            throw new RemoteAccessException("falied to access mdp service [" + methodInvocation.getMethod() + "]", ex);
        }
        try {
            return recreateMdpInvocationResult(result);
        } catch (Throwable ex) {
            if (result.hasInvocationTargetException()) {
                throw ex;
            } else {
                throw new RemoteInvocationFailureException("Error when invoking mdp service ["
                        + methodInvocation.getMethod() + "]", ex);
            }
        }
    }

    /**
     * 发送请求消息执行远程服务
     * 
     * @param invocation
     * @param methodConfig
     * @return
     * @throws JMSException
     */
    protected MdpInvocationResult executeRequest(RemoteInvocation invocation, MdpMethodConfig methodConfig)
            throws MdpException {
        boolean sync = methodConfig.isSync();
        if (sync) {
            return syncRequest(invocation, methodConfig);
        } else {
            return asyncRequest(invocation, methodConfig);
        }
    }

    /**
     * 同步请求
     * 
     * @param invocation
     * @param methodConfig
     * @return
     */
    private MdpInvocationResult syncRequest(RemoteInvocation invocation, MdpMethodConfig methodConfig)
            throws MdpException {
        initIfNecessary();
        if (destinations.length > 1) {
            throw new MdpException("01020030", invocation.getMethodName(), ClassUtils.classNamesToString(invocation
                    .getParameterTypes()));
        }
        long timeout = this.timeout > 0 ? this.timeout : methodConfig.getTimeout();
        long ttl = methodConfig.getTtl();
        int priority = methodConfig.getPriority();
        int deliveryMode = DeliveryMode.PERSISTENT;

        Session session = null;
        MessageProducer producer = null;
        ResponseTub respTub = null;
        Long correlationID = null;
        Lock lock = null;
        Condition condition = null;
        try {
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Message message = createRequestMessage(session, invocation, methodConfig);
            correlationID = correlationIDGenerator.getAndIncrement();
            message.setJMSCorrelationID(Long.toString(correlationID));
            message.setJMSReplyTo(responseQueue);
            Destination destination = resolveDestination(session, destinations[0].trim());
            respTub = new ResponseTub();
            lock = new ReentrantLock();
            condition = lock.newCondition();
            respTub.syncLock = lock;
            respTub.syncCondition = condition;
            responseTubs.put(correlationID, respTub);
            producer = session.createProducer(destination);
            producer.send(message, deliveryMode, priority, ttl);
        } catch (JMSException e) {
            if (correlationID != null) {
                responseTubs.remove(correlationID);
            }
            throw new MdpException("01020055", e, invocation.getMethodName(), ClassUtils.classNamesToString(invocation
                    .getParameterTypes()));
        } finally {
            JmsUtils.closeMessageProducer(producer);
            JmsUtils.closeSession(session);
        }
        if (respTub != null) {
            lock.lock();
            try {
                if (respTub.result == null) {
                    try {
                        condition.await(timeout, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        logger.error("synchronized invocation thread is interrupted.", e);
                    }
                }
                if (respTub.result == null) {
                    responseTubs.remove(correlationID);
                    throw new MdpException("01020050", invocation);
                }
            } finally {
                lock.unlock();
            }
            logger.info("response for [{}]", invocation.getMethodName());
        }
        return respTub.result;
    }

    private MdpInvocationResult asyncRequest(RemoteInvocation invocation, MdpMethodConfig methodConfig)
            throws MdpException {
        initIfNecessary();
        long ttl = methodConfig.getTtl();
        int priority = methodConfig.getPriority();
        int deliveryMode = DeliveryMode.PERSISTENT;
        boolean callback = methodConfig.isCallback();

        if (callback && destinations.length > 1) {
            throw new MdpException("01020065", invocation.getMethodName(), ClassUtils.classNamesToString(invocation
                    .getParameterTypes()));
        }
        Session sessionToClose = null;
        MessageProducer producer = null;
        ResponseTub respTub = null;
        Long correlationID = null;
        MdpCallback targetCallback = null;
        try {
            Session sessionToUse = ConnectionFactoryUtils.doGetTransactionalSession(getConnectionFactory(),
                    new JmsTemplateResourceFactory(transacted), false);
            if (sessionToUse == null) {
                sessionToClose = connection.createSession(transacted, transacted ? Session.SESSION_TRANSACTED
                        : Session.AUTO_ACKNOWLEDGE);
                sessionToUse = sessionToClose;
            }
            targetCallback = replaceCallback(invocation, callback);
            for (int i = 0; i < destinations.length; i++) {
                try {
                    Destination destination = resolveDestination(sessionToUse, destinations[i].trim());
                    Message message = createRequestMessage(sessionToUse, invocation, methodConfig);
                    if (callback) {// 有回调
                        correlationID = correlationIDGenerator.getAndIncrement();
                        message.setJMSCorrelationID(Long.toString(correlationID));
                        message.setJMSReplyTo(responseQueue);
                    }
                    producer = sessionToUse.createProducer(destination);
                    producer.send(message, deliveryMode, priority, ttl);
                    if (targetCallback != null) {
                        respTub = new ResponseTub();
                        respTub.callback = targetCallback;
                        responseTubs.put(correlationID, respTub);
                    }
                } finally {
                    JmsUtils.closeMessageProducer(producer);
                }
            }
            if (sessionToUse.getTransacted() && isSessionLocallyTransacted(sessionToUse, transacted)) {
                JmsUtils.commitIfNecessary(sessionToUse);
            }
            return new MdpInvocationResult(null);
        } catch (JMSException e) {
            throw new MdpException("01020070", invocation.getMethodName(), ClassUtils.classNamesToString(invocation
                    .getParameterTypes()));
        } finally {
            JmsUtils.closeSession(sessionToClose);
        }
    }

    private MdpCallback replaceCallback(RemoteInvocation invocation, boolean callback) {
        MdpCallback targetCallback = null;
        if (callback) {
            Object[] arguments = invocation.getArguments();
            if (arguments != null) {
                for (int i = 0; i < arguments.length; i++) {
                    if (arguments[i] instanceof MdpCallback) {
                        targetCallback = (MdpCallback) arguments[i];
                        arguments[i] = DefaultMdpCallback.INSTANCE;
                        break;
                    }
                }
            }
            invocation.setArguments(arguments);
        }
        return targetCallback;
    }

    /**
     * 处理MDP Service返回的结果
     * 
     * @param message
     * @throws JMSException
     * @throws Throwable
     */
    protected void processResponseMessage(Message message) throws MdpException {
        try {
            String correlationID = message.getJMSCorrelationID();
            if (correlationID == null) {
                onInvalidResponse(message);
                return;
            }
            ResponseTub respTub = responseTubs.remove(Long.valueOf(correlationID));
            if (respTub != null) {
                if (respTub.callback != null) {
                    try {
                        MdpInvocationResult result = extractMdpInvocationResult(message);
                        Object response = result.recreate();
                        logger.info("invoke callback: {}", respTub.callback.getClass().getName());
                        respTub.callback.invoke(response);
                    } catch (Throwable e) {
                        throw new MdpException("01020080", e, message);
                    }
                } else {
                    MdpInvocationResult result = null;
                    try {
                        result = extractMdpInvocationResult(message);
                    } catch (Exception e1) {
                        logger.error(MdpMessages.getMessage("01020020", message), e1);
                    }
                    respTub.syncLock.lock();
                    try {
                        respTub.result = result;
                        respTub.syncCondition.signal();
                    } finally {
                        respTub.syncLock.unlock();
                    }
                }
            } else {
                logger.warn("Invalid response message:{}", message);
            }
        } catch (JMSException e) {
            throw new MdpException("01020040", e);
        }
    }

    protected boolean isSessionLocallyTransacted(Session session, boolean transacted) {
        return transacted && !ConnectionFactoryUtils.isSessionTransactional(session, getConnectionFactory());
    }

    /**
     * 创建请求消息
     * 
     * @param session
     * @param invocation
     * @param methodConfig
     * @return
     * @throws MessageConversionException
     * @throws JMSException
     */
    private Message createRequestMessage(Session session, RemoteInvocation invocation, MdpMethodConfig methodConfig)
            throws MessageConversionException, JMSException {
        Message message = super.createRequestMessage(session, invocation);
        processMessageProperty(message, invocation, methodConfig);
        return message;
    }

    @Override
    public void afterPropertiesSet() {
        initResponseChannel();
    }

    /**
     * 初始化信息返回通道
     */
    private void initResponseChannel() {
        int concurrentConsumers = Integer.parseInt(concurrency);
        sessions = new ArrayList<Session>(concurrentConsumers);
        consumers = new ArrayList<MessageConsumer>(concurrentConsumers);
        try {
            // 增加异常处理监听
            connection = getConnectionFactory().createConnection();
            connection.setExceptionListener(this);
            connection.start();
            for (int i = 0; i < concurrentConsumers; i++) {
                // 处理返回信息不支持事务
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                sessions.add(session);
                if (responseQueue == null) {
                    responseQueue = session.createTemporaryQueue();
                }
                MessageConsumer consumer = session.createConsumer(responseQueue);
                consumer.setMessageListener(new MessageListener() {
                    public void onMessage(final Message message) {
                        try {
                            processResponseMessage(message);
                        } catch (MdpException e) {
                            logger.error(MdpMessages.getMessage("01020040", message), e);
                        }
                    }
                });
                consumers.add(consumer);
            }
        } catch (Throwable e) {
            freeJmsResourceIfError();
            logger.error(MdpMessages.getMessage("01020035"), e);
        }
    }

    private void freeJmsResourceIfError() {
        for (MessageConsumer consumer : consumers) {
            JmsUtils.closeMessageConsumer(consumer);
        }
        consumers = null;
        responseQueue = null;
        for (Session session : sessions) {
            JmsUtils.closeSession(session);
        }
        sessions = null;
        JmsUtils.closeConnection(connection, true);
        connection = null;
    }

    private void initIfNecessary() {
        if (connection != null) {
            return;
        }
        synchronized (this) {
            if (connection != null) {
                return;
            }
            initResponseChannel();
        }
    }

    public void onException(JMSException exception) {
        logger.error("Error on JMS connection", exception);
        freeJmsResourceIfError();
    }

    private Destination resolveDestination(Session session, String destination) throws JMSException {
        return destinationResolver.resolveDestinationName(session, destination, pubSubDomain);
    }

    /**
     * 处理Message Property
     * 
     * @param message
     * @param invocation
     * @param methodConfig
     * @throws JMSException
     */
    private void processMessageProperty(Message message, RemoteInvocation invocation, MdpMethodConfig methodConfig)
            throws JMSException {
        // 处理Argument上的MessageProperty Annotation
        Object[] args = invocation.getArguments();
        if (args != null && args.length > 0) {
            List<List<MessageProperty>> paramAnnotations = methodConfig.getParamAnnotations();
            if (paramAnnotations != null && paramAnnotations.size() > 0) {
                int argIndex = 0;
                for (List<MessageProperty> msgProps : paramAnnotations) {
                    Object arg = args[argIndex];
                    if (msgProps != null && msgProps.size() > 0) {
                        for (MessageProperty msgProp : msgProps) {
                            if (arg == null) {
                                message.setObjectProperty(msgProp.name(), null);
                            } else if (StringUtils.isEmpty(msgProp.jxpath())) {
                                message.setObjectProperty(msgProp.name(), arg);
                            } else {
                                JXPathContext jxpathContext = JXPathContext.newContext(arg);
                                jxpathContext.setLenient(true);
                                message.setObjectProperty(msgProp.name(), jxpathContext.getValue(msgProp.jxpath()));
                            }
                        }
                    }
                    argIndex++;
                }
            }
            // 处理Method上的MessageProperty Annotation
            List<MessageProperty> methodAnnos = methodConfig.getMethodAnnotations();
            if (methodAnnos != null && methodAnnos.size() > 0) {
                for (MessageProperty msgProp : methodAnnos) {
                    message.setObjectProperty(msgProp.name(), msgProp.value());
                }
            }
            // 处理Mdp client 主机名
            message.setStringProperty(MessagePropertiesConstants.MSG_PROPERTY_CLIENT_HOST, getHostName());
            message.setStringProperty(MessagePropertiesConstants.MSG_PROPERTY_CLIENT_IP, getHostIP());
            message.setStringProperty(MessagePropertiesConstants.MSG_PROPERTY_SOURCE, getSource());
        }
    }

    public void setDestinations(String[] destinations) {
        this.destinations = destinations;
    }

    public void setPubSubDomain(boolean pubSubDomain) {
        this.pubSubDomain = pubSubDomain;
    }

    public void setConcurrency(String concurrency) {
        this.concurrency = concurrency;
    }

    public boolean isTransacted() {
        return transacted;
    }

    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    public void setTimeout(long timeout) {
        this.timeout = timeout;
    }

    public void setMdpServiceConfig(MdpServiceConfig mdpServiceConfig) {
        this.mdpServiceConfig = mdpServiceConfig;
    }

    public void setSource(String source) {
        this.source = source;
    }

    public String getSource() {
        if (source == null) {
            source = getHostName();
        }
        return source;
    }

    private String getHostName() {
        if (hostName == null) {
            try {
                hostName = InetAddress.getLocalHost().getHostName();
            } catch (UnknownHostException e) {
                logger.error("Error when geting Mdp client host name", e);
            }
        }
        return hostName;
    }

    private String getHostIP() {
        if (hostIP == null) {
            try {
                hostIP = InetAddress.getLocalHost().getHostAddress();
            } catch (UnknownHostException e) {
                logger.error("Error when geting Mdp client host ip", e);
            }
        }
        return hostIP;
    }

    @Override
    public void setMessageConverter(MessageConverter messageConverter) {
        super.setMessageConverter(messageConverter);
        this.messageConverter = messageConverter;
    }

    public MessageConverter getMessageConverter() {
        return messageConverter;
    }

    protected MdpInvocationResult extractMdpInvocationResult(Message responseMessage) throws JMSException {
        Object content = messageConverter.fromMessage(responseMessage);
        if (content instanceof MdpInvocationResult) {
            return (MdpInvocationResult) content;
        }
        return onInvalidMdpResponse(responseMessage);
    }

    protected Object recreateMdpInvocationResult(MdpInvocationResult result) throws Throwable {
        return result.recreate();
    }

    protected MdpInvocationResult onInvalidMdpResponse(Message responseMessage) throws JMSException {
        throw new MessageFormatException("Invalid response message: " + responseMessage);
    }

    private static class ResponseTub {
        Lock syncLock;
        Condition syncCondition;
        MdpInvocationResult result;
        MdpCallback callback;
    }

    private class JmsTemplateResourceFactory implements ConnectionFactoryUtils.ResourceFactory {
        private final boolean transacted;

        public JmsTemplateResourceFactory(boolean transacted) {
            this.transacted = transacted;
        }

        public Connection getConnection(JmsResourceHolder holder) {
            return holder.getConnection();
        }

        public Session getSession(JmsResourceHolder holder) {
            return holder.getSession();
        }

        public Connection createConnection() throws JMSException {
            return getConnectionFactory().createConnection();
        }

        public Session createSession(Connection con) throws JMSException {
            return con.createSession(transacted, transacted ? Session.SESSION_TRANSACTED : Session.AUTO_ACKNOWLEDGE);
        }

        public boolean isSynchedLocalTransactionAllowed() {
            return transacted;
        }
    }
}
